package tbdp.tas

import java.util.concurrent.TimeUnit

import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.filter.{FilterList, PageFilter}
import org.apache.hadoop.hbase.util.Bytes
import tbdp.tas.model.MinsLine

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
  * Created by xbkaishui on 16/9/11.
  */
class SelectStockHandler(brokers: String, interval: String, dt: String) {

  var dynamicValMap: mutable.Map[String, ArrayBuffer[Double]] = null
  var productFundMap: mutable.Map[String, ArrayBuffer[(Double, Double)]] = null

  /**
    * 从hbase表里面获取分时数据
    *
    * @return
    */
  def loadMinsDataFromHbase(): ArrayBuffer[MinsLine] = {
    val data: ArrayBuffer[MinsLine] = new ArrayBuffer[MinsLine]()
    val table = Utils.initHbaseTable(Constants.hbase_minsline)
    val scan = new Scan()
    val filterList = new FilterList()
    val pageFilter = new PageFilter(100000)
    filterList.addFilter(pageFilter)
    scan.setStartRow(Bytes.toBytes("0_"))
    scan.setFilter(filterList)

    val retryCnt = 3
    for (i <- 1 to retryCnt) {
      //      println(i)
      try {
        val start = System.currentTimeMillis
        val resultScanner = table.getScanner(scan).iterator()
        val end = System.currentTimeMillis
        //        println("scan table cost " + (end - start) + " ms")
        while (resultScanner.hasNext) {
          val r = resultScanner.next()
          try {
            val record: MinsLine = Utils.convert(r, table)
            //只取当天的股票
            if (record.tDate.equalsIgnoreCase(dt)) {
              data.append(record)
            }
          } catch {
            case e: Exception => {
              val key = Bytes.toString(r.getRow())
              println("error convert " + key)
            }
          }
        }
        table.close()
        return data
      } catch {
        case e: Exception => {
          e.printStackTrace()
        }
      }
    }
    //    println("minline hbase data \n" + data)
    table.close()
    data
  }

  def loadDynamicVal(): mutable.Map[String, ArrayBuffer[Double]] = {
    val result = new mutable.HashMap[String, ArrayBuffer[Double]]()
    val sql = "select * from (select sCode ,  marketType , lman_times1 , lman_times4 ,  tdate , row_number() over (partition by SCode,MarketType order by tdate desc) as od  from tbdp.product_dynamicval where tdate < " + dt + " )t where od <=1 "
    println("product_dynamicval sql " + sql)
    val con = Utils.getHiveConn
    // create statement
    val stmt = con.createStatement()
    val res = stmt.executeQuery(sql)
    println(" execute " + sql + " successful")
    while (res.next()) {
      val scode = res.getString(1)
      val marketType = res.getString(2)
      val lman_times1 = res.getDouble(3)
      val lman_times4 = res.getInt(4)
      val data = new ArrayBuffer[Double]()
      data.append(lman_times1)
      data.append(lman_times4)
      result.put(marketType + "_" + scode, data)
    }
    //    println("DynamicVal map " + result)
    result
  }

  def loadProductFund(): mutable.Map[String, ArrayBuffer[(Double, Double)]] = {
    val result = new mutable.HashMap[String, ArrayBuffer[(Double, Double)]]()
    val sql = "select * from (select sCode ,  marketType , allbamt  ,agency1bsvolrate ,  tdate , row_number() over (partition by SCode,MarketType order by tdate desc) as od  from tbdp.product_fund where allbamt >0 and  tdate < " + dt + " )t where od <=5 "
    println("ProductFund sql " + sql)
    val con = Utils.getHiveConn
    // create statement
    val stmt = con.createStatement()
    val res = stmt.executeQuery(sql)
    println(" execute " + sql + " successful")
    while (res.next()) {
      val scode = res.getString(1)
      val marketType = res.getString(2)
      val key = marketType + "_" + scode
      val allbamt = res.getDouble(3)
      val agency1bsvolrate = res.getFloat(4).toDouble
      val data = result.getOrElse(key, new ArrayBuffer[(Double, Double)]())
      data.append((allbamt, agency1bsvolrate))
      result.put(key, data)
    }
    //    println("ProductFund map " + result)
    result
  }

  def getAclose(st: MinsLine): Double = {
    val aClose = {
      if (st.aTrade > 0) {
        st.aTrade
      } else {
        st.yClose
      }
    }
    aClose
  }

  def checkNeedStop(bkMinsLines: ArrayBuffer[MinsLine]) = {
    (bkMinsLines.nonEmpty && bkMinsLines.filter { p => p.tflag == "0" }.isEmpty) || Utils.hqEnd
  }

  def calcMinsLine(): (Iterable[MinsLine], Boolean) = {
    val sts: ArrayBuffer[MinsLine] = loadMinsDataFromHbase()
    val stockMap = sts.map { t => (t.marketType + "_" + t.sCode, t) }.toMap

    val stockList = stockMap.map {
      pair => {
        val key = pair._1
        val st = pair._2
        val close: Double = getAclose(st)
        //计算动态估值
        dynamicValMap.get(key) match {
          case Some(data) => {
            if (close <= (3 * data(1) + data(0)) / 4) {
              st.dynamicVal = 1
            }
            else if ((close > (3 * data(1) + data(0)) / 4) && (close <= (3 * data(0) + data(1)) / 4)) {
              st.dynamicVal = 2
            }
            else if (close > (3 * data(0) + data(1)) / 4) {
              st.dynamicVal = 3
            }
          }
          case _ =>
        }

        //计算 top 资金
        productFundMap.get(key) match {
          case Some(dd) => {
            val data = dd.map(_._1)
            if (st.minsCnt > 0) {
              val day0 = st.allBAmt / st.minsCnt
              if (data.nonEmpty) {
                val day3 = {
                  if (data.length == 3) {
                    data(2)
                  } else {
                    0
                  }
                }
                val day5 = {
                  if (data.length == 5) {
                    data(4)
                  } else {
                    0
                  }
                }
                if (data(0) != 0) {
                  st.topData1 = (day0 - data(0)) * 100 / data(0)
                }
                if (data.take(3).sum != 0) {
                  st.topData3 = (day0 - day3) * 100 / (data.take(3).sum)
                }
                if (data.sum != 0) {
                  st.topData5 = (day0 - day5) * 100 / (data.sum)
                }
              }
            }
          }
          case _ =>
        }

        productFundMap.get(key) match {
          case Some(dd) => {
            val yAgency1BSVolRate = dd.head._2
            st.yAgency1BSVolRate = yAgency1BSVolRate
          }
          case _ =>
        }
        st
      }
    }

    (stockList, checkNeedStop(sts))
  }

  //输出到 mysql
  def sendOut(lines: Iterable[MinsLine]) = {
    val conn = Utils.getMysqlConn
    val sStock = Constants.mysql_select_stock
    val stockSql = s"INSERT INTO $sStock (SCode,MarketType,DynamicVal,TopData1,TopData3,TopData5,AllSco,Agency1BSVolRate,Agency3BSVolRate,Agency5BSVolRate,Agency10BSVolRate,Agency20BSVolRate,YAgency1BSVolRate) " +
      s"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE SCode=? ,MarketType=?, DynamicVal=? ,TopData1=? , TopData3=?,TopData5=?, AllSco=? , Agency1BSVolRate=?,Agency3BSVolRate=?,Agency5BSVolRate=?,Agency10BSVolRate=?,Agency20BSVolRate=?,YAgency1BSVolRate=? ;"
    val stmt = conn.prepareStatement(stockSql)
    var idx = 0
    lines.foreach {
      st => {
        //        println(st.sCode +":"+st.marketType +":\t"+st.agency10BSVolRate)
        idx = idx + 1
        //prepare
        val numList = List(0, 13)
        for (inc <- numList) {
          stmt.setString(1 + inc, st.sCode)
          stmt.setInt(2 + inc, st.marketType.toInt)
          stmt.setInt(3 + inc, st.dynamicVal)
          stmt.setDouble(4 + inc, Utils.formatNumber(st.topData1, 3))
          stmt.setDouble(5 + inc, Utils.formatNumber(st.topData3, 3))
          stmt.setDouble(6 + inc, Utils.formatNumber(st.topData5, 3))
          stmt.setInt(7 + inc, st.allSco)
          stmt.setDouble(8 + inc, Utils.formatNumber(st.agency1BSVolRate, 2))
          stmt.setDouble(9 + inc, Utils.formatNumber(st.agency3BSVolRate, 2))
          stmt.setDouble(10 + inc, Utils.formatNumber(st.agency5BSVolRate, 2))
          stmt.setDouble(11 + inc, Utils.formatNumber(st.agency10BSVolRate, 2))
          stmt.setDouble(12 + inc, Utils.formatNumber(st.agency20BSVolRate, 2))
          stmt.setDouble(13 + inc, Utils.formatNumber(st.yAgency1BSVolRate, 2))
        }
        //        println("stmt sql" + stmt.toString)
        //add batch
        stmt.addBatch()
        //batch execute
        if (idx == 1000) {
          //batch send out
          stmt.executeBatch()
          idx = 0
        }
      }
    }
    if (idx > 0) {
      stmt.executeBatch()
    }
    stmt.close()
    conn.close()
  }

  /**
    * 处理 选股字段生成逻辑
    * 需要计算 动态估值 , top资金1,2,3 , 资金综合分值,机构筹码锁定率_1,3,5,10,20 日
    */
  def handler(): Unit = {
    dynamicValMap = loadDynamicVal()
    productFundMap = loadProductFund()
    val sleepTime = TimeUnit.MINUTES.toMillis(interval.toLong)
    var stop = false
    while (true) {
      try {
        val (bkMinsLines, needStop) = calcMinsLine()
        stop = needStop
        sendOut(bkMinsLines)
        Thread.sleep(sleepTime)
      } catch {
        case e: Error => {
          e.printStackTrace()
        }
      } finally {
        if (stop) {
          println("all stock is end stop process!")
          System.exit(0)
        }
      }
    }
  }

}
